package org.dingfugui.triple.link.zk;

import org.dingfugui.triple.common.Logger;
import org.dingfugui.triple.common.LoggerFactory;
import org.dingfugui.triple.connection.Connection;
import org.dingfugui.triple.connection.ConnectionFactory;
import org.dingfugui.triple.link.Link;
import org.dingfugui.triple.link.RegistrationLink;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 注册中心：
 * 服务端：服务上线，服务下线
 * 客户端：获取服务列表
 *
 * @author fugui.ding at 2015/6/25
 */
public class ZookeeperLink extends RegistrationLink implements Link {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final String appName;
    private ZookeeperWatcher zk;


    public ZookeeperLink(String host, ConnectionFactory factory, String appName) throws IOException {
        super(host, factory);
        this.appName = appName;
        this.factory = factory;
        zk = new ZookeeperWatcher(host, DEFAULT_PORT, appName, rollback);
    }

    public ZookeeperLink(String host, int port, ConnectionFactory factory, String appName) throws IOException {
        super(host, port, factory);
        this.appName = appName;
        zk = new ZookeeperWatcher(host, DEFAULT_PORT, appName, rollback);
    }


    @Override
    public void destory(Connection con) {

    }

    @Override
    public void close() {
        zk.close();
        conMap.values().stream().forEach(list -> list.parallelStream().forEach(con -> con.close()));
    }

    private void fill(Queue<ZookeeperConnection> queue, ConnectionFactory factory, RemoteServerMetaData e) {
        for (int i = 0; i < 10; i++) {
            Connection con = factory.getConnection(new InetSocketAddress(e.getIp(), e.getPort()));
            queue.add(new ZookeeperConnection(con, e));
        }
    }

    private ServerChangeRollback rollback = services -> {
        logger.info("节点变更");
        services.parallelStream().filter(e -> this.appName.equals(e.getAppName())).
                filter(e -> conMap.get(e.getUrl()) == null).forEach(e -> {
            //新增节点
            logger.info("新增节点:{}", e.getUrl());
            Queue<ZookeeperConnection> q = new LinkedBlockingQueue();
            fill(q, factory, e);
            conMap.put(e, q);
        });

        conMap.keySet().parallelStream().filter(e -> !services.contains(e)).forEach(e -> {
            //节点丢失
            logger.info("丢失节点:{}", e.getUrl());
            Queue<ZookeeperConnection> list = conMap.get(e);
            //关闭连接
            list.forEach(con -> con.close());
        });
    };


}
